package com.chaos.eurekaproducer.redis;

import com.alibaba.fastjson.JSON;
import redis.clients.jedis.Jedis;
import com.alibaba.fastjson.TypeReference;
import java.util.Set;
import java.util.UUID;

/**
 * @author liaopeng
 * @title: RedisDelayingQueue
 * @projectName eureka-producer
 * @description: redis实现的延时队列
 * @date 2022/2/13下午4:24
 */
public class RedisDelayingQueue<T> {

    static class TaskItem<T>{
        public String id;
        public T msg;
    }

    private Jedis jedis;

    private String queueKey;

    public RedisDelayingQueue(Jedis jedis, String queueKey) {
        this.jedis = jedis;
        this.queueKey = queueKey;
    }

    public void delay(T msg){
        TaskItem<T> task = new TaskItem<>();
        task.id = UUID.randomUUID().toString();
        task.msg = msg;
        String s = JSON.toJSONString(task);
        jedis.zadd(queueKey,System.currentTimeMillis()+5000,s);
    }

    public void loop(){
        while (!Thread.interrupted()){
            Set<String> values = jedis.zrangeByScore(queueKey, 0, System.currentTimeMillis(), 0, 1);
            if (values.isEmpty()){
                try {
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    break;
                }
                continue;
            }
            String s = values.iterator().next();
            if (jedis.zrem(queueKey,s) > 0){
                TaskItem t = JSON.parseObject(s, new TypeReference<TaskItem>() {});
                System.out.println(t.msg);
            }
        }
    }
}
